package cn.rpc

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}

import scala.collection.mutable
import scala.concurrent.duration._


class Master(val host: String,val port: Int) extends Actor{

  val idToWorker = new mutable.HashMap[String, WorkerInfo]()
  val workers = new mutable.HashSet[WorkerInfo]()
  // 超时检测间隔
  val CHECK_INTERVAL: Long = 15000

  override def preStart(): Unit = {
    println("preStart invoked")
    import context.dispatcher
    context.system.scheduler.schedule(0 millis, CHECK_INTERVAL millis, self, CheckTimeOutWorker)
  }

  // 用于接受消息
  override def receive: Receive = {
    case RegisterWorker(id, memory, cores) => {
      if(!idToWorker.contains(id)){
        val workerInfo = new WorkerInfo(id, memory, cores);
        idToWorker(id) = workerInfo
        workers += workerInfo
        sender ! RegisteredWorker(s"akka.tcp://MasterSystem@$host:$port/user/Master")
      }
    }

    case HeartBeat(id) => {
      if(idToWorker.contains(id)){
        val workerInfo = idToWorker(id)
        //报活
        val currentTime = System.currentTimeMillis()
        workerInfo.lastHeartBeatTime = currentTime
      }
    }

    case CheckTimeOutWorker => {
      val currentTime = System.currentTimeMillis()
      val toRemove = workers.filter(x => currentTime - x.lastHeartBeatTime > CHECK_INTERVAL)
      for(w <- toRemove){
        workers -= w
        idToWorker -= w.id
      }
      println(workers.size)
    }
  }
}

object Master {
  def main(args: Array[String]): Unit = {
    val host = args(0)
    val port = args(1).toInt

    val configStr =
      s"""
        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        |akka.remote.netty.tcp.hostname = "$host"
        |akka.remote.netty.tcp.port = "$port"
      """.stripMargin
    val config = ConfigFactory.parseString(configStr)
    // 单例
    val actorSystem = ActorSystem("MasterSystem", config)
    //创建 actor
    val master = actorSystem.actorOf(Props(new Master(host, port)), "Master")
    master ! "hello"
    actorSystem.awaitTermination()
  }
}
